package com.coolingme.consumer.cyclicsync;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.RandomUtil;
import cn.hutool.core.util.StrUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;

/**
 * CyclicBarrier实现让一组线程等待至某个状态之后再全部同时执行
 *
 * @author wangyue
 * @date 2020/7/8 10:34
 */
@Slf4j
public class CyclicBarrierRunnable implements Runnable {

    public CyclicBarrierRunnable(CyclicBarrier cyclicBarrier, List<ConsumerRecord<?, ?>> recordList) {
        this.cyclicBarrier = cyclicBarrier;
        this.recordList = recordList;
        this.recordListThreadLocal.set(recordList);
    }

    public void setRecordList(List<ConsumerRecord<?, ?>> recordList) {
        this.recordList = recordList;
    }

    private CyclicBarrier cyclicBarrier;

    /**
     * 线程安全的kafka数据列表
     */
    private ThreadLocal<List<ConsumerRecord<?, ?>>> recordListThreadLocal = ThreadLocal.withInitial(() -> new ArrayList<>());

    /**
     * 需要处理的kafka数据，存在线程安全问题
     */
    private List<ConsumerRecord<?, ?>> recordList;

    @Override
    public void run() {
        TimeInterval timerAll = DateUtil.timer();
        TimeInterval timer = DateUtil.timer();
        log.info("{} -正在准备...", Thread.currentThread().getName());
        try {
            for (ConsumerRecord<?, ?> record : recordList) {
                log.info(StrUtil.format("处理数据1：topic= {},partition= {},offset= {},key= {},value= {}",
                        record.topic(), record.partition(), record.offset(), record.key(), record.value()));
            }

            /*for (ConsumerRecord<?, ?> record : recordListThreadLocal.get()) {
                log.info(StrUtil.format("处理数据a：topic= {},partition= {},offset= {},key= {},value= {}",
                        record.topic(), record.partition(), record.offset(), record.key(), record.value()));
            }*/

            // 随机等待,模拟处理时间
            ThreadUtil.sleep(RandomUtil.randomInt(3, 10), TimeUnit.SECONDS);

            /*for (ConsumerRecord<?, ?> record : recordList) {
                log.info(StrUtil.format("处理数据2：topic= {},partition= {},offset= {},key= {},value= {}",
                        record.topic(), record.partition(), record.offset(), record.key(), record.value()));
            }

            for (ConsumerRecord<?, ?> record : recordListThreadLocal.get()) {
                log.info(StrUtil.format("处理数据b：topic= {},partition= {},offset= {},key= {},value= {}",
                        record.topic(), record.partition(), record.offset(), record.key(), record.value()));
            }*/

            log.info("{} -处理完成，等待所有线程完成，耗时：{}", Thread.currentThread().getName(), timer.intervalPretty());
            cyclicBarrier.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
        ThreadUtil.sleep(RandomUtil.randomInt(3, 10), TimeUnit.SECONDS);
        log.info("{} -结束，耗时：{}", Thread.currentThread().getName(), timerAll.intervalPretty());
    }
}
